﻿using System;
using System.Collections.Generic;
using System.Linq;
using Apache.Cassandra;

namespace Sandy
{
    public class MultigetSliceQuery
    {
        public static IDictionary<string, string> Perform(SandyClient client,
                    SandyFamily family, IEnumerable<string> keys, string column)
        {
            var parent = new ColumnParent { Column_family = family.Name };

            SlicePredicate pred = new SlicePredicate
            {
                Column_names = new List<byte[]>() { family.ColumnNameToBytes(column) }
            };

            var res = client.Client.multiget_slice(client.Keyspace, keys.ToList(), parent, pred, ConsistencyLevel.ONE);

            // Prepare our format
            var dict = new OrderPreservingDictionary<string, string>();
            foreach (var kv in res.Where(kv => kv.Value.Count > 0))
            {
                ColumnOrSuperColumn col = kv.Value.FirstOrDefault();
                if (col == null) continue;

                dict.Add(kv.Key, family.GetValueString(col.Column.Value));
            }

            return dict;
        }

        public static IDictionary<string, string> Perform(SandyClient client,
            SandyFamily family, IEnumerable<string> keys, string superCol, string column)
        {
            var parent = new ColumnParent
                {
                    Column_family = family.Name,
                    Super_column = family.SuperColumnNameToBytes(superCol)
                };

            SlicePredicate pred = new SlicePredicate
            {
                Column_names = new List<byte[]>() { family.ColumnNameToBytes(column) }
            };

            var res = client.Client.multiget_slice(client.Keyspace, keys.ToList(), parent, pred, ConsistencyLevel.ONE);

            // Prepare our format
            var dict = new OrderPreservingDictionary<string, string>();
            foreach (var kv in res.Where(kv => kv.Value.Count > 0))
            {
                ColumnOrSuperColumn scol = kv.Value.FirstOrDefault();
                if (scol == null) continue;

                dict.Add(kv.Key, family.GetValueString(scol.Column.Value));
            }

            return dict;

        }

        public static IDictionary<string, IDictionary<string, string>> PerformGetAll(SandyClient client,
            SandyFamily family, IEnumerable<string> keys, string superCol)
        {
            var parent = new ColumnParent
            {
                Column_family = family.Name
            };

            SlicePredicate pred = new SlicePredicate
            {
                Column_names = new List<byte[]>() { family.SuperColumnNameToBytes(superCol) }
            };

            var res = client.Client.multiget_slice(client.Keyspace, keys.ToList(), parent, pred, ConsistencyLevel.ONE);

            // Prepare our format
            var dict = new OrderPreservingDictionary<string, IDictionary<string, string>>();
            foreach (var kv in res.Where(kv => kv.Value.Count > 0))
            {
                dict.Add(kv.Key, family.CreateSupercolDict(kv.Value.FirstOrDefault().Super_column));
            }

            return dict;

        }

        private static IDictionary<string, IDictionary<string, string>> PerformInternal(SandyClient client,
            SandyFamily family, IEnumerable<string> keys, ColumnParent parent, SlicePredicate pred)
        {
            var res = client.Client.multiget_slice(client.Keyspace, keys.ToList(), parent, pred, ConsistencyLevel.ONE);

            // Prepare our format
            var dict = new OrderPreservingDictionary<string, IDictionary<string, string>>();
            foreach (var kv in res.Where(kv => kv.Value.Count > 0))
            {
                dict.Add(kv.Key, family.CreateColumnDict(kv.Value));
            }

            return dict;
        }

        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
            SandyFamily family, IEnumerable<string> keys, string superCol, IEnumerable<string> cols)
        {
            var parent = new ColumnParent
            {
                Column_family = family.Name,
                Super_column = family.SuperColumnNameToBytes(superCol)
            };

            var pred = new SlicePredicate
            {
                Column_names = cols.Select(family.ColumnNameToBytes).ToList()
            };

            return PerformInternal(client, family, keys, parent, pred);
        }

        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
            SandyFamily family, IEnumerable<string> keys, string superCol, Range colRange)
        {
            var parent = new ColumnParent
            {
                Column_family = family.Name,
                Super_column = family.SuperColumnNameToBytes(superCol)
            };

            var pred = new SlicePredicate
            {
                Slice_range = colRange.ToSliceRange(family)
            };

            return PerformInternal(client, family, keys, parent, pred);
        }


        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
            SandyFamily family, IEnumerable<string> keys, IEnumerable<string> cols)
        {
            var parent = new ColumnParent { Column_family = family.Name };

            SlicePredicate pred = new SlicePredicate
                                        {
                                            Column_names = cols.Select(family.ColumnNameToBytes).ToList()
                                        };

            return PerformInternal(client, family, keys, parent, pred);
        }

        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
            SandyFamily family, IEnumerable<string> keys, Range colRange)
        {
            var parent = new ColumnParent { Column_family = family.Name };

            SlicePredicate pred = new SlicePredicate
            {
                Slice_range = colRange.ToSliceRange(family)
            };

            return PerformInternal(client, family, keys, parent, pred);
        }

        public static IDictionary<string, IDictionary<string, IDictionary<string, string>>>
            PerformSuper(SandyClient client, SandyFamily family, IEnumerable<string> keys, IEnumerable<string> superCols)
        {
            SlicePredicate pred = new SlicePredicate
            {
                Column_names = superCols.Select(family.SuperColumnNameToBytes).ToList()
            };

            return PerformSuperInternal(client, family, keys, pred);
        }

        public static IDictionary<string, IDictionary<string, IDictionary<string, string>>>
            PerformSuper(SandyClient client, SandyFamily family, IEnumerable<string> keys, Range scolRange)
        {
            SlicePredicate pred = new SlicePredicate
            {
                Slice_range = scolRange.ToSuperSliceRange(family)
            };

            return PerformSuperInternal(client, family, keys, pred);
        }

        private static IDictionary<string, IDictionary<string, IDictionary<string, string>>>
            PerformSuperInternal(SandyClient client, SandyFamily family, IEnumerable<string> keys, SlicePredicate pred)
        {
            var parent = new ColumnParent { Column_family = family.Name };

            var res = client.Client.multiget_slice(client.Keyspace, keys.ToList(), parent, pred, ConsistencyLevel.ONE);

            // Prepare our result dict (k -> (scn -> (cn -> cv)))
            var dict = new OrderPreservingDictionary<string, IDictionary<string, IDictionary<string, string>>>();
            foreach (var kvpair in res)
            {
                var scDict = new OrderPreservingDictionary<string, IDictionary<string, string>>();
                foreach (var scol in kvpair.Value.Select(c => c.Super_column))
                {
                    scDict.Add(family.SuperColumnNameToString(scol.Name), family.CreateSupercolDict(scol));
                }
                dict.Add(kvpair.Key, scDict);
            }

            return dict;
        }
    }
}